package virtualinit;

import com.alibaba.fastjson.annotation.JSONField;
import virtualinit.util.CommonUtil;
import virtualinit.util.JdbcUtil;
import virtualinit.util.Log;
import virtualinit.util.SSHUtil;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;

public class Ds {
    static Log log = Log.instance;
    static Config config = Config.instance;
    public transient LinkedBlockingQueue<Connection> connections = new LinkedBlockingQueue<>(1);
    public int dsSize;
    public String id;
    public String name;
    public String url;
    public String username;
    public String password;

    public List<Database> databases = new ArrayList<>();

    public void validate() {
        CommonUtil.assertIsNull("datasource id=null", id);
        CommonUtil.assertIsNull("datasource name=null", name);
        CommonUtil.assertIsNull("datasource url=null", url);
        CommonUtil.assertIsNull("datasource username=null", username);
        CommonUtil.assertIsNull("datasource password=null", password);
    }

    public void init(Integer size) {
        validate();
        long start = System.currentTimeMillis();
        if (size != null) dsSize = size;

        log.info("准备初始化datasource:{} size:{}", name, dsSize);

        if (SSHUtil.hasSession()) {
            String hostPort = JdbcUtil.getHostPortFromUrl(url);
            String localPort = name.contains("schedule") ? "9527" : "9528";
            this.url = url.replace(hostPort, "localhost:" + localPort);
            log.info("开启ssh，mysql链接ip需要替换，将{}：{} 替换为localhost:{}", name, hostPort, localPort);
        }

        connections = new LinkedBlockingQueue<>(dsSize);
        for (int i = 0; i < dsSize; i++) {
            try {
                connections.put(JdbcUtil.getConnection(url, username, password));
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        log.info("初始化datasource:{} size:{} 完成，耗时：{}ms", name, dsSize, System.currentTimeMillis() - start);
    }

    public void initAllDsInfos() {
        Connection conn = connections.peek();
        // 查询所有db和 tables
        this.databases = JdbcUtil.executeToMap(conn, "show databases").stream().flatMap(e -> e.values().stream())
                .filter(e -> e.toString().toLowerCase().startsWith(config.wastonDbPrefix.toLowerCase()))
                .map(e -> new Database((String) e))
                .collect(Collectors.toList());

        this.databases.forEach(database -> {
            ResultSet rs = null;
            try {
                rs = conn.getMetaData().getTables(database.databaseName, null, null, null);
                JdbcUtil.rsToMap(rs).stream()
                        .forEach(e -> {
                            String table_name = e.get("TABLE_NAME").toString();
                            database.tables.add(new Table(table_name, database.databaseName));
                        });
            } catch (SQLException ex) {
                ex.printStackTrace();
            } finally {
                CommonUtil.closeQuietly(rs);
            }
        });

        log.info("\n数据库和数据源映射关系:{} 详细信息：\n{}", this.name, Config.prettyJson(databases));
    }


    public Connection takeConnection() {
        try {
            return connections.take();
        } catch (InterruptedException e) {
            log.error("take connection thread Interrupted", e);
            return null;
        }
    }

    public Connection peekConnection() {
        return connections.peek();
    }

    public void returnConnection(Connection conn) {
        if (conn != null) {
            try {
                connections.put(conn);
            } catch (InterruptedException e) {
                log.error("return connection thread Interrupted", e);
            }
        }
    }


    public void close() {
        log.info("closing Ds[{}]. size:{}", name, connections.size());
        for (Connection connection : connections) {
            CommonUtil.closeQuietly(true, connection);
        }
        connections.clear();
    }

    public static Log getLog() {
        return log;
    }

    public static void setLog(Log log) {
        Ds.log = log;
    }

    public int getDsSize() {
        return dsSize;
    }

    public void setDsSize(int dsSize) {
        this.dsSize = dsSize;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }


    public static class Database {
        public String databaseName;
        @JSONField(serializeUsing = DefaultSerialize.class)
        public List<Table> tables = new ArrayList<>();

        public Database(String databaseName) {
            this.databaseName = databaseName;
        }
    }

    public static class Table {
        public String tableName;
        public String databaseName;

        public Table(String tableName, String databaseName) {
            this.tableName = tableName;
            this.databaseName = databaseName;
        }
    }
}
